package com.example.bigdata.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import scala.actors.threadpool.Arrays;

import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ConsumerRecordThreadSample {
    public final static String TOPIC_NAME = "hrsjw1_topic";

    /**
     *  consumer 非阻塞形式异步处理数据，consumer 接到数据后 交给后面的handler 进行异步数据处理
     *  consumer 不能处理offset 用于非业务型数据处理
     */
    public static void main(String[] args) throws InterruptedException {
        String brokerList ="localhost:9092";
        String groupId = "test";
        int workerNum = 5 ;
        ConsumerExecutor consumer = new ConsumerExecutor(brokerList,groupId,TOPIC_NAME);
        consumer.execute(workerNum);
        Thread.sleep(1000000);
        consumer.shutdown();
    }

    // Consumer线程池
    public static class ConsumerExecutor {
        private final KafkaConsumer<String,String> consumer;
        private ExecutorService executors ;
        public ConsumerExecutor(String brokerList,String groupId,String topic){
            Properties props = new Properties();
            props.setProperty("bootstrap.servers", brokerList);
            props.setProperty("group.id", groupId);
            props.setProperty("enable.auto.commit", "false");
            props.setProperty("auto.commit.interval.ms", "1000");
            props.setProperty("session.timeout.ms", "30000");
            props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumer = new KafkaConsumer(props);
            consumer.subscribe(Arrays.asList(new String[]{topic}));
        }
        public void execute(int workerNum) {
            executors = new ThreadPoolExecutor(workerNum,workerNum,0L, TimeUnit.MILLISECONDS,
                    new ArrayBlockingQueue<>(1000),new ThreadPoolExecutor.CallerRunsPolicy());
            while (true){
                ConsumerRecords<String, String> records = consumer.poll(200);
                for(final ConsumerRecord record:records){
                    executors.submit(new ConsumerRecordWorker(record));
                }
            }
        }
        public void shutdown(){
            if(consumer !=null){
                consumer.close();
            }
            if(executors != null){
                executors.shutdown();
            }
            try {
                if(!executors.awaitTermination(10,TimeUnit.SECONDS)){
                    System.out.println("Timeout ... Ignore for this case");
                }
            }catch (InterruptedException ignored){
                System.out.println("Other thread Interrupted this shutdown.Ignore for this case");
                Thread.currentThread().interrupt();
            }
        }

        // 处理记录
        public static class ConsumerRecordWorker implements Runnable {
            private ConsumerRecord<String,String> record;
            public ConsumerRecordWorker(ConsumerRecord record){this.record = record;}

            @Override
            public void run(){
                System.out.println("Thread - "+Thread.currentThread().getName());
                System.err.printf("partition = %d , offset = %d , key = %s , value = %s%n",
                        record.partition(),record.offset(),record.key(),record.value());
            }
        }

    }
}
